package kafka_consume_optimize

import (
	"context"
	"encoding/json"
	"fmt"
	"hash/fnv"
	"log"
	"strings"
	"sync"
	"time"

	"github.com/segmentio/kafka-go"
	"go-demo-2025/common"
	"go-demo-2025/utils/kafkaUtil"
)

// kafka 消费逻辑优化
// 测试用例: tests/kafka_tests/kafka_consume_optimize_test.go

// 订单消息结构体
type OrderMessage struct {
	OrderID string `json:"order_id"`
	Amount  int64  `json:"amount"`
	Status  string `json:"status"`
}

// 优化后的Kafka生产者（带分区路由）
type OptimizedProducer struct {
	writer       *kafka.Writer
	partitionCnt int // 分区总数
}

// 优化后的Kafka消费者（多分区并发消费）
type OptimizedConsumer struct {
	readers   []*kafka.Reader // 每个分区一个reader
	wg        sync.WaitGroup
	ctx       context.Context
	cancel    context.CancelFunc
	batchSize int // 批量处理大小
}

// 初始化生产者（指定分区数）
func NewOptimizedProducer(topic string, partitionCnt int) (*OptimizedProducer, error) {
	if partitionCnt <= 0 {
		return nil, fmt.Errorf("分区数必须大于0")
	}

	writer := &kafka.Writer{
		Addr:     kafka.TCP(kafkaUtil.KafkaBrokers...),
		Topic:    topic,
		Balancer: &kafka.LeastBytes{},
	}

	return &OptimizedProducer{
		writer:       writer,
		partitionCnt: partitionCnt,
	}, nil
}

// 发送消息（按OrderID路由到固定分区）
func (p *OptimizedProducer) SendOrderMessage(ctx context.Context, order *OrderMessage) error {
	if order == nil || order.OrderID == "" {
		return fmt.Errorf("订单信息不能为空")
	}

	// 序列化消息
	value, err := json.Marshal(order)
	if err != nil {
		return fmt.Errorf("序列化失败: %v", err)
	}

	// 计算目标分区
	partition := hashOrderIDToPartition(order.OrderID, p.partitionCnt)

	// 发送到指定分区
	err = p.writer.WriteMessages(ctx, kafka.Message{
		Key:       []byte(order.OrderID),
		Value:     value,
		Partition: partition,
	})
	if err != nil {
		return fmt.Errorf("发送消息失败: %v", err)
	}

	log.Printf("消息发送成功: order_id=%s, 分区=%d", order.OrderID, partition)
	return nil
}

// 关闭生产者
func (p *OptimizedProducer) Close() error {
	return p.writer.Close()
}

// 初始化消费者（消费所有分区）
func NewOptimizedConsumer(topic string, groupID string, partitionCnt int, batchSize int) (*OptimizedConsumer, error) {
	if partitionCnt <= 0 {
		return nil, fmt.Errorf("分区数必须大于0")
	}
	if batchSize <= 0 {
		return nil, fmt.Errorf("批量大小必须大于0")
	}

	ctx, cancel := context.WithCancel(context.Background())
	readers := make([]*kafka.Reader, 0, partitionCnt)

	// 为每个分区创建独立reader（指定分区时不能有GroupID）
	for i := 0; i < partitionCnt; i++ {
		reader := kafka.NewReader(kafka.ReaderConfig{
			Brokers:   kafkaUtil.KafkaBrokers,
			Topic:     topic,
			Partition: i, // 固定消费指定分区
			MinBytes:  1024,
			MaxBytes:  1024 * 1024,
		})
		readers = append(readers, reader)
	}

	return &OptimizedConsumer{
		readers:   readers,
		ctx:       ctx,
		cancel:    cancel,
		batchSize: batchSize,
	}, nil
}

// 启动消费者（每个分区独立协程消费）
func (c *OptimizedConsumer) Start() {
	log.Printf("启动优化版消费者，分区数=%d，批量大小=%d", len(c.readers), c.batchSize)

	for i, reader := range c.readers {
		c.wg.Add(1)
		go c.consumePartition(i, reader)
	}
}

// 消费指定分区消息（批量处理）
func (c *OptimizedConsumer) consumePartition(partition int, reader *kafka.Reader) {
	defer c.wg.Done()
	batch := make([]kafka.Message, 0, c.batchSize)
	ticker := time.NewTicker(5 * time.Second) // 批量超时定时器
	defer ticker.Stop()

	log.Printf("开始消费分区%d", partition)

	for {
		select {
		case <-c.ctx.Done():
			log.Printf("分区%d消费协程退出", partition)
			// 处理剩余消息
			if len(batch) > 0 {
				if err := c.processBatch(batch); err != nil {
					log.Printf("分区%d批量处理剩余消息失败: %v", partition, err)
				} else {
					log.Printf("分区%d成功处理剩余%d条消息", partition, len(batch))
				}
				_ = c.commitBatch(reader, batch)
			}
			return

		case <-ticker.C:
			// 超时处理批量消息
			if len(batch) > 0 {
				log.Printf("分区%d批量超时，处理%d条消息", partition, len(batch))
				if err := c.processBatch(batch); err != nil {
					log.Printf("分区%d批量处理失败: %v", partition, err)
				} else {
					_ = c.commitBatch(reader, batch)
				}
				batch = batch[:0]
			}

		default:
			// 读取消息
			msg, err := reader.ReadMessage(c.ctx)
			if err != nil {
				log.Printf("分区%d读取消息失败: %v，1秒后重试", partition, err)
				time.Sleep(1 * time.Second)
				continue
			}

			// 加入批量队列
			batch = append(batch, msg)
			if len(batch) >= c.batchSize {
				log.Printf("分区%d达到批量阈值，处理%d条消息", partition, len(batch))
				if err := c.processBatch(batch); err != nil {
					log.Printf("分区%d批量处理失败: %v", partition, err)
				} else {
					_ = c.commitBatch(reader, batch)
				}
				batch = batch[:0]
				ticker.Reset(5 * time.Second)
			}
		}
	}
}

// 批量处理消息（无分布式锁）
func (c *OptimizedConsumer) processBatch(msgs []kafka.Message) error {
	orders := make([]*OrderMessage, 0, len(msgs))
	orderIDs := make([]string, 0, len(msgs))

	// 解析消息
	for _, msg := range msgs {
		var order OrderMessage
		if err := json.Unmarshal(msg.Value, &order); err != nil {
			return fmt.Errorf("解析消息失败: %v, key=%s", err, string(msg.Key))
		}
		orders = append(orders, &order)
		orderIDs = append(orderIDs, order.OrderID)
	}

	// 批量更新订单状态（核心优化：无锁批量处理）
	return batchUpdateOrderStatus(c.ctx, orders)
}

// 批量提交偏移量
func (c *OptimizedConsumer) commitBatch(reader *kafka.Reader, msgs []kafka.Message) error {
	lastMsg := msgs[len(msgs)-1]
	if err := reader.CommitMessages(c.ctx, lastMsg); err != nil {
		return fmt.Errorf("提交偏移量失败: %v", err)
	}
	log.Printf("成功提交偏移量: partition=%d, offset=%d", lastMsg.Partition, lastMsg.Offset)
	return nil
}

// 停止消费者
func (c *OptimizedConsumer) Stop() {
	log.Println("开始停止消费者...")
	c.cancel()
	c.wg.Wait()
	for _, reader := range c.readers {
		_ = reader.Close()
	}
	log.Println("消费者已完全停止")
}

// 哈希函数：将OrderID映射到指定分区
func hashOrderIDToPartition(orderID string, partitionCount int) int {
	h := fnv.New32a()
	_, _ = h.Write([]byte(orderID))
	return int(h.Sum32() % uint32(partitionCount))
}

// 批量更新订单状态（数据库操作）
func batchUpdateOrderStatus(ctx context.Context, orders []*OrderMessage) error {
	if len(orders) == 0 {
		return nil
	}

	// 构建批量SQL
	placeholders := make([]string, 0, len(orders))
	args := make([]interface{}, 0, len(orders)*2)

	for _, order := range orders {
		placeholders = append(placeholders, "(?, ?)")
		args = append(args, order.OrderID, "paid")
	}

	sql := fmt.Sprintf(`
		INSERT INTO orders (order_id, amount, status, create_time, update_time) 
		VALUES %s
		ON DUPLICATE KEY UPDATE 
		status = VALUES(status), 
		amount = VALUES(amount),
		update_time = DEFAULT
	`, strings.Join(placeholders, ","))

	// 执行批量操作
	_, err := common.SqlDB.ExecContext(ctx, sql, args...)
	if err != nil {
		return fmt.Errorf("批量更新失败: %v", err)
	}

	log.Printf("成功批量处理%d条订单: %v", len(orders), orders)
	return nil
}

/*备注:orders表结构:
CREATE TABLE IF NOT EXISTS orders (
    order_id VARCHAR(64) PRIMARY KEY,
    amount BIGINT NOT NULL,
    status VARCHAR(32) NOT NULL,
    create_time DATETIME DEFAULT CURRENT_TIMESTAMP,
    update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);
*/
